package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo3flatMap {
  def main(args: Array[String]): Unit = {
    /**
     * flatMap算子：转换算子
     * 需要接收一个函数f：参数类型同RDD中的泛型，返回值类型是数据容器（集合、数组、序列、迭代器）
     * 会将每一条数据依次传递个函数f进行转换，还会将函数f返回的数据容器进行扁平化处理（展开）
     * 最终整个flatMap方法完成后会返回一个新的RDD
     */
    // 构建Spark上下文环境
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo3flatMap")
    conf.setMaster("local")
    val sc: SparkContext = new SparkContext(conf)

    val lineRDD: RDD[String] = sc.parallelize(List("java,java,scala,python", "hadoop,hive,hbase", "spark,flink,MapReduce"))

    lineRDD.foreach(println)

    val wordsRDD: RDD[String] = lineRDD.flatMap(line => line.split(","))

    wordsRDD.foreach(println)

  }

}
